/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.loading;

import java.nio.ByteBuffer;
import java.util.logging.Logger;

import edu.ucsd.db.bassarisk.data.api.IValueType.IValueReference;
import edu.ucsd.db.bassarisk.data.compression.VariableLengthUnsignedLongSequenceCodec;
import edu.ucsd.db.bassarisk.data.values.LongType;
import edu.ucsd.db.bassarisk.data.values.LongType.LongTools;
import edu.ucsd.db.bassarisk.data.values.LongType.LongValueCodec;
import edu.ucsd.db.bassarisk.data.values.LongType.ObjectLongReference;
import edu.ucsd.db.bassarisk.storage.common.ColumnStore;

/**
 * Card Layout
 * 
 * <ul>
 * <li>| 8 bytes (last value)
 * <li>| V bytes (first value)
 * <li>| V bytes (delta first to second value)
 * <li>| V bytes (delta first to third value)
 * <li>...
 * <li>| V bytes (delta n-2'th value to n-1'th value)
 * </ul>
 */
public class DeltaEncodedAscendingUnsignedLongCardLoader implements
    ICardLoader<LongType> {
  private static final Logger LOG = 
      Logger.getLogger(DeltaEncodedAscendingUnsignedLongCardLoader.class.getName());

  private enum LoaderState {
    Invalid,
    Empty,
    Filling,
    Full
  }

  private final VariableLengthUnsignedLongSequenceCodec vlongCodec;

  private final LongValueCodec longCodec;
  private final ByteBuffer cardBuffer;

  private final ColumnSegmentInfo<LongType> cardInfo;
  private long previousValue = -1;

  private LoaderState loaderState = LoaderState.Invalid;

  public DeltaEncodedAscendingUnsignedLongCardLoader() {
    super();
    this.vlongCodec = VariableLengthUnsignedLongSequenceCodec.INSTANCE;
    this.longCodec = LongTools.INSTANCE.getGenericCodec();

    this.cardBuffer = ByteBuffer.allocate(ColumnStore.BYTES_IN_CARD);
    this.cardInfo = new ColumnSegmentInfo<LongType>(LongTools.INSTANCE);
  }

  @Override
  public boolean appendBufferedValue(final IValueReference<LongType> value) {
    assert this.loaderState != LoaderState.Invalid;

    if (this.loaderState == LoaderState.Empty) {
      this.cardInfo.setFirstValue(value);
      this.loaderState = LoaderState.Filling;
    } else if (this.loaderState == LoaderState.Full) {
      this.loaderState = LoaderState.Invalid;
      return false;
    }

    long longValue = this.longCodec.getLongValue(value);
    long longDelta = longValue - this.previousValue;

    assert longDelta >= 0;

    int numBytes = this.vlongCodec.encode(this.cardBuffer, longDelta);
    if (numBytes == -1) {
      this.writeFinal(longValue);
    }

    this.previousValue = longValue;
    this.cardInfo.incrementEntryCount();
    return true;
  }

  @Override
  public boolean fileCard(final DrawerLoader<LongType> drawer) {
    if (this.cardInfo.getEntryCount() == 0) { return true; }
    this.writeFinal(this.previousValue);
    // Succeeds if-and-only-if drawer append succeeds.
    return drawer.appendCard(this.cardBuffer.array(), this.cardInfo);
  }

  @Override
  public byte[] getCardBytes() {
    return this.cardBuffer.array();
  }

  @Override
  public void initialize() {
    this.reset(0);
  }

  @Override
  public void reset(final int cardEntryStart) {
    this.loaderState = LoaderState.Empty;
    this.cardInfo.reset(cardEntryStart);
    this.cardBuffer.clear();
    this.previousValue = 0; // delta from zero to first value is first value
    this.cardBuffer.position(8); // first 8 bytes are for final value
  }

  private void writeFinal(final long value) {
    this.cardInfo.setLastValue(new ObjectLongReference(value));
    this.cardBuffer.putLong(0, value);
    this.loaderState = LoaderState.Full;
  }

}
